﻿
#include "Base_MessageCore.h"
#include "Base_Monitor.h"
#include "Base_MessageQueue.h"
#include "Base_Context.h"
#include <assert.h>
#include <iostream>

#define _DEBUG

#include "Log.h"

bool Base_MessageCore::start(const Config & conf)
{
	m_monitor = Base_Monitor::get_instance();
	m_gmq = Base_GlobalQueue::get_global_queue_instance();
	int thread_work_count = conf.thread_work_count;
	m_start = true;
	m_thread_arr= new std::thread*[thread_work_count];
	for (int i=0;i<thread_work_count;i++)
	{
		std::thread * t = new std::thread{std::bind(&Base_MessageCore::worker,this)};
		m_thread_arr[i] = t;
	}

	LOG("消息核心开始工作\n");   

	return true;
}

void Base_MessageCore::quit()
{
	for (int i=0;i<m_thread_work_count;i++)
	{
		delete m_thread_arr[i];
	}
	delete [] m_thread_arr;
}

Base_MessageCore * Base_MessageCore::get_instance()
{
	static Base_MessageCore c;
	return &c;
}


void Base_MessageCore::send_msg(uint32_t type, uint32_t src, uint32_t des, void * data, int size)
{
	Base_Context * ctx = nullptr;
	auto f = m_map_context.find(des);
	assert(f != m_map_context.end());
	ctx = f->second;
	ctx->get_mq()->mq_push(Base_Message(des, src, 0, data, type, size));
	DEBUG("投递消息\n");
	m_monitor->weak_up();
	DEBUG("唤醒\n");
}

void Base_MessageCore::register_context(Base_Context *context)
{
	uint32_t h = context->get_handle();
	auto f=m_map_context.find(h);
	if (f==m_map_context.end())
	{
		m_map_context.emplace(std::make_pair(h, context));
	}
}

Base_MessageCore::Base_MessageCore()
	:m_gmq(nullptr)
	, m_monitor(nullptr)
	, m_start(false)
	, m_thread_arr(nullptr)
	,m_thread_work_count(0)
{

}

Base_MessageCore::~Base_MessageCore()
{

}

void Base_MessageCore::worker()
{

	Base_Message msg;
	int s;
	while (m_start)
	{
		DEBUG("正在运行core thread_id:%d\n",std::this_thread::get_id());
		Base_MessageQueue* q = m_gmq->global_queue_pop();
		if (q!=nullptr)
		{
			do 
			{
				s=q->mq_pop(msg);
				DEBUG("处理一条消息\n剩余消息：%d\n",s);
				q->get_context()->callback(&msg);
			} while (s<0);
		}
		else
		{
			std::unique_lock <std::mutex> m{ m_monitor->mutex };
			m_monitor->thread_work_state = Base_Monitor::wait;
			m_monitor->condition.wait(m);
			std::cout << std::this_thread::get_id() << std::endl;
		}

	}
}

std::map<uint32_t, Base_Context *> Base_MessageCore::m_map_context;

